/*
 * webqueue.c
 *
 *  Created on: Sep 12, 2010
 *      Author: thienlong
 */

#include <stddef.h>
#include <time.h>
#include <event2/buffer.h>
#include <ctype.h>
#include <event2/keyvalq_struct.h>
#include <sys/queue.h>
#include <assert.h>
#include <string.h>
#include <arpa/inet.h>
#include <stdlib.h>

#include "server.h"
#include "stringbuffer.h"
#include "server.h"
#include "map.h"
#include "log.h"
#include <event2/http_struct.h>
#include <errno.h>
#include "filecache.h"
#include "http.h"
#include <sys/stat.h>
#include <dirent.h>
#include <sys/inotify.h>
#include "cache.h"
#include <stdio.h>
#include "file_notify.h"

const int SOCKET_ADRR_SIZE = sizeof(struct sockaddr_storage);
extern lru_cache *file_cache;
extern ServerConf *conf;
#ifdef HAVE_SERVER_STATUS
extern struct server_status status;
#endif
void* thread_status_handler(void *args);

void do_accept(int listener, short event_flags, void *args) {
	struct thread_state *thread = thread_get_request_handler(
			conf->httpThreadNum);
	struct socket_item *sck = lpool_get_element(&thread->cache,
			SOCKET_ITEM_TYPE);
	if (!sck) {
		sck = socket_item_new();
	}
	sck->socklen = SOCKET_ADRR_SIZE;
	sck->fd = accept(listener, (struct sockaddr*) &sck->ss, &sck->socklen);
	if (sck->fd < 0) {
		lpool_put_element(&thread->cache, SOCKET_ITEM_TYPE,
				(struct next_data*) sck);
		time_t t = time(NULL);
		char buf[64];
		int errsv = errno;
		fprintf_log(LOG_ERROR, "[webqueue.c] accept() failed at %s due %s",
				ctime_r(&t, buf), strerror(errsv));
		return;
	}
#ifdef HAVE_SERVER_STATUS
	server_status_increase_conn(&status);
#endif
	evutil_make_socket_nonblocking(sck->fd);
	dispatch_connection(sck, thread);
}

void response_content_file(stringbuffer *content, struct evhttp_request *req) {
	struct evbuffer *buf = evhttp_request_get_output_buffer(req);
	if (buf == NULL) {
		fprintf_log(LOG_ERROR,
				"[%s] In %s() %d: failed to create response buffer", __FILE__,
				__FUNCTION__, __LINE__);
	}
	evbuffer_add(buf, content->ptr, content->used);
	char *extension = file_get_extension(req->uri + 1, -1);
	char *content_type = NULL;
	if (extension != NULL && (content_type = getHeaderByMimeType(
			conf->httpMimeTypes, extension)) != NULL) {
		evhttp_add_header(req->output_headers, CONTENT_TYPE_HEADER,
				content_type);
	}
}

val load_content_file(key k, mem_storage *storage, void *arg) {
	char *name = k;
	char *webroot = arg;
	char filename[256];
	strcpy(filename, webroot);
	strcat(filename, name);
	FILE * file = fopen(filename, "r");
	int flag = -1;
	stringbuffer *buffer = NULL;
	if (file) {

		fseek(file, 0, SEEK_END);
		int length = ftell(file);
		fseek(file, 0, SEEK_SET);
		buffer = stringbuffer_new();
		stringbuffer_prepare_copy(buffer, length);
		if (length != 0 && fread(buffer->ptr, length, 1, file) < 1) {
			fprintf_log(LOG_ERROR, "Can't read file: %s ", filename);
		} else {
			flag = 0;
		}
		buffer->used = length;
		fclose(file);
	}
	return flag == 0? buffer : NULL ;
}

void server_handler_default(struct evhttp_request *req, void *args) {
	struct thread_state *thread = args;
#ifdef HAVE_SERVER_STATUS
	server_status_increase_req(&status);
#endif
	stringbuffer *content = lru_cache_get(file_cache, req->uri, thread->thread_id);
	if (content != NULL && content->used != 0) {
		response_content_file(content, req);
		evhttp_send_reply(req, HTTP_OK, "OK", NULL);
	} else {
		evbuffer_add(req->output_buffer, "NOT FOUND", strlen("NOT FOUND"));
		evhttp_send_reply(req, HTTP_NOTFOUND, "NOT FOUND", NULL);
	}
	//	evbuffer_add_printf(buf, "{\"now\":%d}", (int) time(NULL));

}

void server_status(struct evhttp_request * req, void *args) {
	struct thread_state *thread = args;
	linked_pool *cache = &thread->cache;
	server_status_increase_req(&status);

	struct evkeyvalq *params = (struct evkeyvalq*) lpool_get_element(
			cache, EVKEYVALUE_TYPE);
	if (!params) {
		params = (struct evkeyvalq*) malloc(sizeof(struct evkeyvalq));
	}

	if (req->uri) {
		evhttp_parse_query(req->uri, params);
	}
	char *interval_str = evhttp_find_header(params, INTERVAL_PARAM);
	if (interval_str) {
		unsigned int *interval = malloc(sizeof(int));
		*interval = atoi(interval_str);
		pthread_t thread_id;
		pthread_attr_t thread_attr;
		pthread_attr_init(&thread_attr);

		if (pthread_create(&thread_id, &thread_attr, thread_status_handler,
				interval) != 0) {
			fprintf_log(LOG_ERROR,
					"[%s] In %s() %d: can't create new status handler thread",
					__FILE__, __FUNCTION__, __LINE__);
		}
		goto reply_response;
	}
	stringbuffer *buf = lpool_get_element(cache, STRING_BUFFER_TYPE);
	if (!buf) {
		buf = stringbuffer_new();
	}
	stringbuffer_append_string(buf, "server's start time:\t");
	time_t startTime = server_status_get_starttime(&status);
	stringbuffer_append_string(buf, ctime(&startTime));
	stringbuffer_append_string(buf, "<br>total connections number:\t");
	long connNum = server_status_get_conn(&status);
	stringbuffer_append_long(buf, connNum);
	stringbuffer_append_string(buf, "<br>total request number:\t");
	long reqNum = server_status_get_req(&status);
	stringbuffer_append_long(buf, reqNum);
	stringbuffer_append_string(buf, "<br>thread_id:\t");
	stringbuffer_append_long(buf, thread->thread_id);
	stringbuffer_append_string(buf, "<br>data_cache's status:\t");
	lpool_to_string(cache, buf);
	stringbuffer_append_string(buf, "<br>size of pending socket:\t");
	stringbuffer_append_long(buf, socket_queue_size(&thread->connect_sockets));
	struct evbuffer *output_buffer = evhttp_request_get_output_buffer(req);
	if (output_buffer == NULL) {
		fprintf_log(LOG_ERROR,
				"[%s] In %s() %d: failed to create response buffer", __FILE__,
				__FUNCTION__, __LINE__);
	}
	evbuffer_add(output_buffer, buf->ptr, buf->used - 1);
	lpool_put_element(cache, STRING_BUFFER_TYPE, (struct next_data*) buf);
	//reply http response
	reply_response: evhttp_send_reply(req, HTTP_OK, "OK", NULL);
	evhttp_clear_headers(params);
	lpool_put_element(cache, EVKEYVALUE_TYPE, (struct next_data*) params);
}

void* thread_status_handler(void *args) {
	//guarantees that thread resources are deallocated upon return
	pthread_detach(pthread_self());
	unsigned int interval = *((unsigned int *) args);
	if (interval > 0) {
		stringbuffer *buf = stringbuffer_new();
		stringbuffer_append_string(buf, "server's start time:\t");
		time_t startTime = server_status_get_starttime(&status);
		stringbuffer_append_string(buf, ctime(&startTime));
		stringbuffer_append_string(buf, "\ntotal connections number:\t");
		long connNum = server_status_get_conn(&status);
		stringbuffer_append_long(buf, connNum);
		stringbuffer_append_string(buf, "\ntotal request number:\t");
		long reqNum = server_status_get_req(&status);
		stringbuffer_append_long(buf, reqNum);
		sleep(interval);
		connNum = server_status_get_conn(&status) - connNum;
		reqNum = server_status_get_req(&status) - reqNum;
		fprintf_log(
				LOG_INFO,
				"%s\nInterval: %ds has %d connections\t%d requests\n\t%.2f conn/s\t%.2f freq/s",
				buf->ptr, interval, connNum, reqNum,
				connNum / (float) interval, reqNum / (float) interval);
		stringbuffer_free(buf);
	}
	free(args);
	return NULL;
}

void *file_checker_handler(void *arg) {
	//guarantees that thread resources are deallocated upon return
	pthread_detach(pthread_self());
	ServerConf *conf = (ServerConf *) arg;
	stringbuffer *webroot = stringbuffer_new_string(conf->httpWebRoot);
	if (webroot->ptr[webroot->used - 1] != '/') {
		stringbuffer_append_char(webroot, '/');
	}

	struct event_base *base = event_base_new();
	file_notify *notify = file_notify_new(webroot, NULL, IN_CREATE | IN_MODIFY
			| IN_DELETE | IN_MOVE, 1);
	struct event *listen_event = event_new(base, notify->listener, EV_READ
			| EV_PERSIST, notify_handler, notify);
	if (event_add(listen_event, NULL) < 0) {
		perror("add event");
		exit(1);
	}
	event_base_dispatch(base);
	return NULL;
}
#define EVENT_SIZE  ( sizeof (struct inotify_event) )
#define EVENT_BUF_LEN     ( 1024 * ( EVENT_SIZE + 16 ) )

char buffer[EVENT_BUF_LEN];

void notify_handler(int listener, short event_flags, void *args) {
	int length = 0;
	file_notify *notify = (file_notify *) args;
	int counter_read = 0;
	int counter_event = 0;
	char *moved_dirname = NULL;
	int moved_wd = -1;
	while (1) {
		length = read(listener, buffer, EVENT_BUF_LEN);
		++counter_read;
		if (length <= 0) {
			assert(errno == EAGAIN);
			if (moved_dirname) {
				file_notify_rm_watch_dirname_recursive(notify, moved_wd,
						moved_dirname);
			}
			break;
		}
		int i = 0;
		while (i < length) {
			struct inotify_event *event = (struct inotify_event *) &buffer[i];
			++counter_event;
			i += EVENT_SIZE + event->len;
			if (event->len) {
				if (event->mask & IN_MOVED_FROM) {
					printf("File or Directory %s has move from\n", event->name);
				}

				if (event->mask & IN_MOVED_TO) {
					printf("File or Directory %s has move to\n", event->name);
				}
				if (event->mask & IN_ISDIR) {
					if (moved_dirname && !(event->mask & IN_MOVED_TO)) {
						file_notify_rm_watch_dirname_recursive(notify,
								moved_wd, moved_dirname);
						moved_dirname = NULL;
					}
					if (event->mask & IN_DELETE) {
						file_notify_rm_watch_dirname(notify, event->wd,
								event->name);
						printf("Directory %s delete.\n", event->name);
					} else if (event->mask & IN_MODIFY) {
						printf("Directory %s modify.\n", event->name);
					} else if (event->mask & IN_CREATE) {
						printf("New directory %s created.\n", event->name);
						int res = file_notify_add_watch(notify, event->wd,
								event->name, notify->flags);
						if (res == -1) {
							fprintf_log(LOG_ERROR, "can't watch directory: %s",
									event->name);
						}
						continue;
					} else if (!moved_dirname && (event->mask & IN_MOVED_TO)) {
						printf("New directory %s was move to.\n", event->name);
						int res = file_notify_add_watch_recursive(notify,
								event->wd, event->name, notify->flags);
						if (res == -1) {
							fprintf_log(LOG_ERROR, "can't watch directory: %s",
									event->name);
						}
						continue;
					} else if (event->mask & IN_MOVED_FROM) {
						moved_dirname = event->name;
						moved_wd = event->wd;
					} // IN_MOVED_FROM
					else if (event->mask & IN_MOVED_TO) {
						file_notify_replace_wath_dirname(notify, moved_wd,
								moved_dirname, event->wd, event->name);
						moved_dirname = NULL;
					}

				} else {
					stringbuffer *filepath = lpool_get_element(
							notify->cache, STRING_BUFFER_TYPE);
					if (filepath == NULL) {
						filepath = stringbuffer_new();
					}
					stringbuffer_copy_stringbuffer(filepath,
							notify->wd_dir[event->wd]);
					stringbuffer_append_string(filepath, event->name);
					if (event->mask & IN_DELETE) {
//						lru_cache_remove_char(file_cache, filepath->ptr
//								+ notify->root_dir->used - 1);
						printf("File %s delete.\n", event->name);
					} else if (event->mask & IN_MODIFY) {
						printf("File %s modify.\n", event->name);
					}
					lpool_put_element(notify->cache, STRING_BUFFER_TYPE,
							(struct next_data*) filepath);
				}
			}
		}
	}
	printf("counter of read: %d\tcounter of event: %d\n", counter_read,
			counter_event);
	fflush(stdout);
}

char* file_get_extension(char *filename, int ex_max_len) {
	return file_get_extension_len(filename, strlen(filename), ex_max_len);
}

char* file_get_extension_len(char *filename, int str_len, int ex_max_len) {
	char *end = filename + str_len - 1;

	char *start = filename;
	if (ex_max_len != -1 && str_len > ex_max_len) {
		start = end - ex_max_len;
	}
	while (end != start) {
		if (*end == '.') {
			return end + 1;
		}
		--end;
	}
	return NULL;
}

